package com.wxzhou.wangzhuan.service;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wxzhou.wangzhuan.dto.AdReportInfo;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.hash.ReactiveHashCommands;
import io.quarkus.redis.datasource.hash.ReactiveHashScanCursor;
import io.quarkus.redis.datasource.keys.ReactiveKeyCommands;
import io.quarkus.redis.datasource.stream.ReactiveStreamCommands;
import io.quarkus.redis.datasource.value.ReactiveValueCommands;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 广告上报数据发布到redis服务
 */
@ApplicationScoped
public class AdLogPublishService {

    private static final org.jboss.logging.Logger LOG = Logger.getLogger(AdLogPublishService.class);
    private static final String AD_ID_SN_PREFIX = "ad:idsn:";

    //最大保留24小时
    private static final long DEFAULT_EXPIRE = Duration.ofHours(24).getSeconds();

    private final ReactiveStreamCommands<String, String, String> streamCommands;
    private final ReactiveValueCommands<String, String> valueCommands;
    private final ReactiveKeyCommands<String> keyCommands;
    private final ReactiveHashCommands<String, String, String> hashCommands;


    private final ObjectMapper objectMapper;

    @Inject
    public AdLogPublishService(ReactiveRedisDataSource redisDataSource, ObjectMapper objectMapper) {
        this.streamCommands = redisDataSource.stream(String.class, String.class, String.class);
        this.valueCommands = redisDataSource.value(String.class);
        this.hashCommands = redisDataSource.hash(String.class, String.class, String.class);
        keyCommands = redisDataSource.key(String.class);
        this.objectMapper = objectMapper;
        this.objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL).configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    /**
     * 存储data到redis中,自动分组
     *
     * @param data   广告上报数据列表
     * @param userid 用户ID
     * @return Uni<Boolean> 存储是否成功
     */
    public Uni<Boolean> storeAdInfoBy(List<AdReportInfo> data, Long userid) {
        if (data == null || data.isEmpty() || userid == null) {
            LOG.warn("Invalid parameters for storing ad info");
            return Uni.createFrom().item(false);
        }

        try {
            // 按ad_id_sn和sense进行分组
            Map<String, Map<Integer, List<AdReportInfo>>> groupedData = data.stream()
                    .collect(Collectors.groupingBy(
                            AdReportInfo::getAd_id_sn,
                            Collectors.groupingBy(AdReportInfo::getSense)
                    ));

            // 创建所有的存储操作
            List<Uni<?>> operations = new ArrayList<>();

            // 遍历每个ad_id_sn分组
            for (Map.Entry<String, Map<Integer, List<AdReportInfo>>> adIdSnEntry : groupedData.entrySet()) {
                String adIdSn = adIdSnEntry.getKey();

                Map<Integer, List<AdReportInfo>> senseMap = adIdSnEntry.getValue();

                // 为每个sense创建hash字段
                for (Map.Entry<Integer, List<AdReportInfo>> senseEntry : senseMap.entrySet()) {
                    Integer sense = senseEntry.getKey();
                    String redisKey = buildAdIdSnKey(userid); // 格式：ad:idsn:{adIdSn}:{userid}

                    // 按bs_ac值再次分组
                    Map<Integer, List<AdReportInfo>> bsAcMap = senseEntry.getValue().stream()
                            .collect(Collectors.groupingBy(AdReportInfo::getBs_ac));

                    HashSet<Integer> bsacs = new HashSet<>(bsAcMap.keySet());
                    //若bs_ac=0,从hash中,找到某个同filed,读取数据,比如读取出来是[1],bsacs=[2,3],转换成Set,更新该filed

                    String field = adIdSn + "_" + sense;

                    operations.add(
                            findAllByAdidsns(adIdSn, sense, userid).onItem().transform(oldBsacs -> {
                                bsacs.addAll(oldBsacs);
                                // 添加存储操作
                                String jsonData = null;
                                try {
                                    jsonData = objectMapper.writeValueAsString(bsacs);
                                } catch (JsonProcessingException e) {
                                    throw new RuntimeException(e);
                                }
                                return hashCommands.hset(redisKey, field, jsonData)
                                        .subscribe().
                                        with(
                                                aBoolean -> LOG.info("写入成功"),
                                                throwable -> LOG.errorf(throwable, "Error storing ad info"));
                            })
                    );

                    // 设置过期时间
                    operations.add(keyCommands.expire(redisKey, DEFAULT_EXPIRE)
                            .onFailure().invoke(e -> {
                                LOG.error("Error setting expiration for key: " + redisKey, e);
                            })
                    );
                }
            }

            // 并行执行所有操作
            return Uni.combine().all().unis(operations)
                    .discardItems()
                    .replaceWith(true)
                    .onFailure().recoverWithItem(false);
        } catch (Exception e) {
            LOG.error("Error storing ad info", e);
            return Uni.createFrom().failure(e);
        }
    }

    public Uni<Map<String, Set<Integer>>> findMapByAdidsn(Set<String> thisAdidsns,Long userid) {
        //从redis中批量获取AD_ID_SN_PREFIX+adidsn,如:ad:idsn:*,用hscan模糊查找AD_ID_SN_PREFIX+":*"然后再过滤adidsns中包含的项,返回所有匹配项的所有数据
        return Uni.createFrom().voidItem()
                .onItem().transformToUni(unused -> {
                    // 执行hscan命令
                    ReactiveHashScanCursor<String, String> hscan = hashCommands.hscan(buildAdIdSnKey(userid));
                    return hscan.toMulti().collect().asList()
                            .onItem().transformToUni(hashFields -> {
                                // 过滤出adidsns中包含的项
                                Map<String, Set<Integer>> resultByAdidsn = new HashMap<>();
                                // 遍历所有的hash字段
                                for (Map.Entry<String, String> entry : hashFields) {
                                    String fieldKey = entry.getKey();//示例:a_1,表示adidsn是a,sense是1
                                    String jsonValue = entry.getValue();
                                    try {
                                        // 解析fieldKey获取adidsn和sense
                                        String[] parts = fieldKey.split("_");
                                        if (parts.length >= 2) {
                                            String adidsn = parts[0];
                                            if (thisAdidsns.contains(adidsn)) {
                                                // 解析jsonValue获取bs_ac集合
                                                Set<Integer> adReportInfos = objectMapper.readValue(jsonValue, objectMapper.getTypeFactory().constructCollectionType(Set.class, Integer.class));

                                                // 将数据添加到结果Map中
                                                resultByAdidsn.computeIfAbsent(adidsn, k -> new HashSet<>()).addAll(adReportInfos);
                                            }
                                        }
                                    } catch (Exception e) {
                                        LOG.error("Failed to process ad report data for field: " + fieldKey, e);
                                    }
                                }
                                return Uni.createFrom().item(resultByAdidsn);
                            });
                })
                .onFailure().recoverWithItem(Collections.emptyMap());
    }

    /**
     * 根据adidsn，批量获取
     *
     * @param adidsn
     * @return
     */
    public Uni<Set<Integer>> findAllByAdidsns(String adidsn, int sense, Long userid) {
        //从redis中批量获取AD_ID_SN_PREFIX+adidsn,如:ad:idsn:*,用hscan模糊查找AD_ID_SN_PREFIX+":*"然后再过滤adidsns中包含的项,返回所有匹配项的所有数据
        return Uni.createFrom().voidItem()
                .onItem().transformToUni(unused -> {
                    // 执行hscan命令
                    ReactiveHashScanCursor<String, String> hscan = hashCommands.hscan(buildAdIdSnKey(userid));
                    return hscan.toMulti().collect().asList()
                            .onItem().transformToUni(hashFields -> {
                                // 过滤出adidsns中包含的项
                                Set<Integer> allReports = new HashSet<>();
                                // 遍历所有的hash字段
                                for (Map.Entry<String, String> entry : hashFields) {
                                    String fieldKey = entry.getKey();
                                    String jsonValue = entry.getValue();
                                    try {
                                        // 解析fieldKey获取adidsn和sense
                                        String[] parts = fieldKey.split("_");
                                        if (parts.length >= 2 && parts[0].equals(adidsn) && parts[1].equals(String.valueOf(sense))) {
                                            Set<Integer> adReportInfos = objectMapper.readValue(jsonValue, objectMapper.getTypeFactory().constructCollectionType(Set.class, Integer.class));
                                            allReports.addAll(adReportInfos);
                                            break;
                                        }
                                    } catch (Exception e) {
                                        LOG.error("Failed to process ad report data for field: " + fieldKey, e);
                                    }
                                }
                                return Uni.createFrom().item(allReports);
                            });
                })
                .onFailure().recoverWithItem(Collections.emptySet());
    }


    /**
     * 构建adIdSn对应的Redis键
     */
    private String buildAdIdSnKey(Long userid) {
        LocalDateTime now = LocalDateTime.now();
        String format = now.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        return AD_ID_SN_PREFIX + format + ":" + userid;
    }
}
